package cn._51doit.live.jobs;

import cn._51doit.live.pojo.DataBean;
import cn._51doit.live.udfs.IsNewUserFunction;
import cn._51doit.live.udfs.JsonToBeanFunction;
import cn._51doit.live.utils.EventType;
import cn._51doit.live.utils.FlinkUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
 * 实时统计统计新老用户（从今天凌晨开始，到当前）
 * 数据中没有isNew这个字段了，需要根据设备ID进行计算
 */
public class NewUserCountV2 {

    public static void main(String[] args) throws Exception {

        //从指定的文件中读取参数
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(args[0]);

        DataStream<String> lines = FlinkUtils.createKafkaStream(parameterTool, SimpleStringSchema.class);

        //将数据进行转换（字符串转成Bean）
        SingleOutputStreamOperator<DataBean> beanStream = lines.process(new JsonToBeanFunction());

        //过滤数据类型
        SingleOutputStreamOperator<DataBean> filtered = beanStream.filter(bean -> EventType.APP_LAUNCH.equals(bean.getEventId()));

        //按照【设备类型】进行KeyBy，希望设备ID相同的数据不论怎样，数据一定要进入到同一个分区中
        //存在的问题：可能存在数据倾斜（某一些设备类型的用户比较多，有的比较少）
        KeyedStream<DataBean, String> keyedStream = filtered.keyBy(DataBean::getDeviceType);

        SingleOutputStreamOperator<DataBean> res = keyedStream.process(new IsNewUserFunction());

        res.print();

        FlinkUtils.env.execute();

    }
}
